package com.ry.flink.function;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
/**
 * 最终输出：
 *  2022-11-20 22:10:55 type area count
 * */
public class IndicatorProcessMainByTransportTypeApplySumFun implements WindowFunction<Tuple3<Long,String,String>, Tuple4<String,String,String,Long>, Tuple, TimeWindow> {
    static SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


    /**
     * 结果聚合
     * */
    @Override
    public void apply(Tuple key, TimeWindow window, java.lang.Iterable<Tuple3<Long, String, String>> input, Collector<Tuple4<String, String, String, Long>> out) {
        String area = key.getField(0).toString();
        String type = key.getField(1).toString();

        //次数
        long count = 0;
        Iterator<Tuple3<Long, String, String>> iterator = input.iterator();
        while (iterator.hasNext()) {
            iterator.next();
            count++;
        }
        //结果输出
        String time = IndicatorProcessMainByTransportTypeApplySumFun.format.format(new Date(window.getEnd()));
        out.collect(Tuple4.of(time,area,type,count));
    }

}
